package com.adwyxx.collection;

import java.util.*;
import java.util.concurrent.*;

/**
 * {@link Queue} 接口：作为 {@link Collection}的子接口，Queue 定义了队列数据结构，队列是一种先进先出（FIFO）的数据结构，即先进入队列的元素会先被取出。
 * <pre>
 *  poll()                                   offer(N)
 *           ----------------------------------
 *   出队 <--  [A]  [B]  [C]  [D]  [E] ... [N]  <-- 入队
 *           ----------------------------------
 *             ↑                            ↑
 *           first                        last
 * </pre>
 * Queue 集合常用的实现类：
 *  - 双端队列的两个实现类：LinkedList、ArrayDeque
 *  - 优先级队列：PriorityQueue
 *  - 线程安全相关的Queue实现类：LinkedBlockingQueue、ArrayBlockingQueue、ConcurrentLinkedQueue
 *
 * {@link Deque} 双端队列: Deque接口是Queue接口的子接口，它定义了双端队列的数据结构，允许在两端插入和移除元素。
 * <pre>
 *  pollFirst()                                   offerLast(N)
 *  出队 ↖ --------------------------------- ↙ 入队
 *         [A]  [B]  [C]  [D]  [E] ... [N]
 *  入队 ↗ --------------------------------- ↘ 出队
 *  offerFirst(A)                                 pollLast()
 * </pre>
 * Deque 的实现类：
 *  - 双端队列的两个实现类：LinkedList、ArrayDeque
 *
 * <pre>
 *                                                |-> [ArrayDeque]
 *                                 |-> [Deque<E>] |-> [LinkedList]
 *                                 |              |-> [BlockingDeque]-> [LinkedBlockingDeque]
 * [Iterable<T>]                   |
 *      ↓                          |                   |-> [PriorityQueue]
 * [Collection<E>] -> [Queue<E>] ->|-> [AbstractQueue] |-> [ConcurrentLinkedQueue]
 *                                 |
 *                                 |                   |-> [ArrayBlockingQueue]
 *                                 |                   |-> [LinkedBlockingQueue]
 *                                 |-> [BlockingQueue] |-> [PriorityBlockingQueue]
 *                                                     |-> [DelayQueue]
 *                                                     |-> [LinkedTransferQueue]
 *                                                     |-> [SynchronousQueue]
 *
 * </pre>
 *
 * @author Leo.W
 * @date 2024-09-20 9:19
 */
public class QueueDemo {
    public static void main(String[] args) {
        // queueMethod();
        producerConsumer();
        concurrentLinkedQueueTest();
    }

    /**
     * Queue接口常用方法:
     *  - add(E e) : 将元素e插入到队列尾部，如果插入成功返回true，如果队列已满，抛出IllegalStateException异常
     *  - offer(E e) : 将元素e插入到队列尾部，如果插入成功返回true，如果队列已满，返回false
     *  - remove() : 移除并返回队列头部的元素，如果队列为空，抛出NoSuchElementException异常
     *  - poll() : 移除并返回队列头部的元素，如果队列为空，返回null
     *  - element() : 返回队列头部的元素，如果队列为空，抛出NoSuchElementException异常
     *  - peek() : 返回队列头部的元素，如果队列为空，返回null
     *  - size() : 返回队列中元素的个数
     *  - isEmpty() : 判断队列是否为空
     *  - iterator() : 返回队列的迭代器
     *  - toArray() : 返回队列的数组形式
     *  - toArray(T[] a) : 返回队列的数组形式，数组类型为T
     *  - clear() : 清空队列
     *  - contains(Object o) : 判断队列中是否包含元素o
     *  - remove(Object o) : 从队列中移除元素o
     */
    public static void queueMethod() {
        Queue<String> queue = new PriorityQueue<>();

        // add(E e) : 将元素e插入到队列尾部，如果插入成功返回true，如果队列已满，抛出IllegalStateException异常
        queue.add("a");
        queue.add("b");
        queue.add("c");

        // offer(E e)：将元素插入到队尾
        System.out.println("queue.offer(\"d\") -> "+queue.offer("d")+ ", then -> "+queue);

        System.out.println(queue);

        // poll(): 移除并返回队列头部的元素，如果队列为空，返回null
        System.out.println("queue.poll() -> " + queue.poll() +" , aflter poll() -> "+queue);

        // remove()：移除并返回队列头部的元素，如果队列为空，抛出NoSuchElementException异常
        System.out.println("queue.remove() -> " + queue.remove() +" , aflter remove() -> "+queue);

        // peek(): 返回队列头部的元素，注意：不会删除元素，如果队列为空，返回null
        System.out.println("queue.peek() -> " + queue.peek() +" , aflter peek() -> "+queue);

        // element(): 返回队列头部的元素，注意：不会删除元素，如果队列为空，抛出NoSuchElementException异常
        System.out.println("queue.element() -> " + queue.element() +" , aflter element() -> "+queue);

        // iterator 遍历队列
        Iterator<String> iterator =  queue.iterator();
        while (iterator.hasNext()) {
            System.out.println(iterator.next());
        }

        // for each 遍历
        for (String item : queue) {
            System.out.println(item);
        }

        // forEach 遍历
        queue.forEach(System.out::println);

        System.out.println("queue.size() -> "+queue.size());
        System.out.println("queue.isEmpty() -> "+queue.isEmpty());

        // queue.toArray();
        // queue.toArray(new String[0]);
        queue.clear();
        System.out.println("queue.contains(\"a\") -> "+queue.contains("a"));
        System.out.println("queue.remove(\"a\") -> "+queue.remove("a"));
    }

    /**
     * 线程安全的阻塞队列 BlockingQueue 应用典型： 生产者-消费者 模型
     * Java提供的线程安全的Queue可以分为 阻塞队列 和 非阻塞队列：
     *  - 阻塞队列的典型例子是 BlockingQueue 接口，其常用的实现类有ArrayBlockingQueue、LinkedBlockingQueue 、PriorityBlockingQueue等。
     *      + 阻塞队列的特点是当队列为空时，消费者线程会被阻塞，直到队列中有元素；当队列满时，生产者线程会被阻塞，直到队列中有空位。
     *      + 阻塞队列通常用于生产者-消费者模式中，生产者线程将数据放入队列，消费者线程从队列中取出数据进行处理。
     *      + 阻塞队列的典型应用场景是线程池、消息队列等。
     *      + ArrayBlockingQueue : 其构造函数必须带一个int参数来指明其大小
     *      + LinkedBlockingQueue : 若其构造函数带一个规定大小的参数，生成的BlockingQueue有大小限制，若不带大小参数，所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定
     *      + PriorityBlockingQueue : 其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序
     *  - 非阻塞队列的典型例子是 ConcurrentLinkedQueue
     *      + 非阻塞队列的特点是当队列为空时，消费者线程不会被阻塞，而是返回null或者抛出NoSuchElementException异常；
     *      + 当队列满时，生产者线程不会被阻塞，而是返回false或者抛出IllegalStateException异常。
     *      + 非阻塞队列通常用于需要快速响应的场景，例如网络通信、游戏开发等。
     *
     * BlockingQueue:
     *  + 有界或无界队列：BlockingQueue 可以是有界的，也可以是无界的，但无界队列在某些实现（如 LinkedBlockingQueue）中可能会因为内存不足而抛出 OutOfMemoryError。
     *  + 阻塞操作：BlockingQueue 提供了一些阻塞方法，如 put() 和 take()，当队列满时，put() 方法会阻塞直到有空间可用；当队列为空时，take() 方法会阻塞直到有元素可用。这使得 BlockingQueue 更适合于生产者-消费者模式。
     *  + 线程安全：BlockingQueue 也是线程安全的，多个线程可以并发地对其进行操作，而不会出现数据不一致的问题。
     *  + 适用于`生产者-消费者模式`：由于 BlockingQueue 提供了阻塞操作，它非常适合用于实现生产者-消费者模式，其中生产者线程将元素放入队列，消费者线程从队列中取出元素。
     */
    public static void producerConsumer() {
        // 创建容量为10的线程安全的阻塞队列
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(10);
        // 生产者线程
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    queue.put(i); // 添加元素到队列，如果队列满，线程将阻塞等待
                    System.out.println(Thread.currentThread().getName() + " put  -> " + i);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // 重新中断线程
                    e.printStackTrace();
                }
            }
        },"Producer");

        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                // 循环等待
                while (true) {
                    int item = queue.take(); // 从队列取出元素，如果队列为空，线程将阻塞等待
                    System.out.println(Thread.currentThread().getName() + " take -> " + item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 重新中断线程
                e.printStackTrace();
            }
        },"Consumer");

        producer.start();
        consumer.start();
    }

    /**
     * 线程安全的非阻塞队列 ConcurrentLinkedQueue 应用典型： 生产者-消费者 模型
     * ConcurrentLinkedQueue 是 Java 中 java.util.concurrent 包提供的一个线程安全的无界非阻塞队列。
     * 它实现了 Queue 接口，并使用 CAS（Compare-And-Swap）操作来保证线程安全。
     * ConcurrentLinkedQueue 适用于高并发环境下的多线程操作，因为它在大多数情况下不需要锁，因此性能较好。
     * ConcurrentLinkedQueue 的主要特性：
     *  + 无界队列：ConcurrentLinkedQueue 是一个无界队列，这意味着它不会因为容量问题而阻塞。
     *  + 线程安全：ConcurrentLinkedQueue 是线程安全的，多个线程可以并发地对其进行操作，而不会出现数据不一致的问题。
     *  + 非阻塞：ConcurrentLinkedQueue 使用 CAS 操作来实现非阻塞的并发访问，因此在大多数情况下，它不会因为锁竞争而导致性能下降。
     *  + 适用于高并发环境：由于 ConcurrentLinkedQueue 的非阻塞特性，它非常适合在需要处理大量并发操作的场景中使用。
     */
    private static void concurrentLinkedQueueTest(){
        ConcurrentLinkedQueue<Integer> concurrentLinkedQueue = new ConcurrentLinkedQueue<>();
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        // 生产者线程
        executorService.submit(new Producer("producer1",concurrentLinkedQueue));
        executorService.submit(new Producer("producer2",concurrentLinkedQueue));
        executorService.submit(new Producer("producer3",concurrentLinkedQueue));
        // 消费者线程
        executorService.submit(new Consumer("consumer1",concurrentLinkedQueue));
        executorService.submit(new Consumer("consumer2",concurrentLinkedQueue));
        executorService.submit(new Consumer("consumer3",concurrentLinkedQueue));
    }

    /**
     * 生产者线程定义
     */
    static class Producer implements Runnable {
        private String name;
        private  ConcurrentLinkedQueue<Integer> concurrentLinkedQueue;

        public Producer(String name, ConcurrentLinkedQueue<Integer> queue) {
            this.name = name;
            this.concurrentLinkedQueue = queue;
        }

        public void run() {
            for (int i = 1; i < 10; ++i) {
                System.out.println(name+ " start producer " + i);
                concurrentLinkedQueue.add(i);
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //System.out.println(name+"end producer " + i);
            }
        }
    }

    /**
     * 消费者线程定义
     */
    static class Consumer implements Runnable {
        private String name;
        private  ConcurrentLinkedQueue<Integer> concurrentLinkedQueue;

        public Consumer(String name, ConcurrentLinkedQueue<Integer> queue) {
            this.name = name;
            this.concurrentLinkedQueue = queue;
        }
        public void run() {
            for (int i = 1; i < 10; ++i) {
                try {
                    System.out.println(name+" Consumer " +  concurrentLinkedQueue.poll());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
